package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.HbaseUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @Author lzc
 * @Date 2023/8/12 11:48
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {
    
    private RedisClient asyncRedisClient;
    private StatefulRedisConnection<String, String> asyncRedisConnection;
    private AsyncConnection asyncHbaseConnection;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取一个Redis异步客户端
        asyncRedisClient = RedisUtil.getAsyncRedisClient();
        asyncRedisConnection = RedisUtil.getAsyncRedisConnection(asyncRedisClient);
        
        // 获取一个异步的 Hbase 连接
        asyncHbaseConnection = HbaseUtil.getAsyncConnection();
    }
    
    @Override
    public void close() throws Exception {
        
        if (asyncRedisConnection != null) {
            asyncRedisConnection.close();
        }
        // 关闭redis异步的客户端
        if (asyncRedisClient != null) {
            asyncRedisClient.shutdown();
        }
        
        // 关闭 hbase 的异步连接
        HbaseUtil.closeAsyncConnection(asyncHbaseConnection);
        
    }
    
    @Override
    public void asyncInvoke(T bean,
                            ResultFuture<T> resultFuture) throws Exception {
        // 用到 java8 的异步框架
        CompletableFuture
            .<JSONObject>supplyAsync(new Supplier<JSONObject>() {
                @Override
                public JSONObject get() {
                    // 异步的去读取 redis . 返回值就是从 redis 读取到的维度数据
                    return RedisUtil.asyncGetOneRow(asyncRedisConnection,
                                                    getTableName(),
                                                    getId(bean),
                                                    JSONObject.class
                    );
                }
            })
            .thenApplyAsync(new Function<JSONObject, JSONObject>() {
                // 异步的去读取 hbase
                @Override
                public JSONObject apply(JSONObject dim) {
                    if (dim == null) {
                        System.out.println("走的 hbase: " + getTableName() + "   " + getId(bean));
                        dim = HbaseUtil.asyncGetOneRow(asyncHbaseConnection,
                                                       "gmall",
                                                       getTableName(),
                                                       getId(bean),
                                                       JSONObject.class
                        );
                        
                        // 把查询到的维度写入到 Redis 中
                        RedisUtil.asyncWriteOneRow(asyncRedisConnection,
                                                   getTableName(),
                                                   getId(bean),
                                                   dim
                        );
                    }else{
                        System.out.println("走的 redis: " + getTableName() + "   " + getId(bean));
                    }
                    return dim;
                }
            })
            .thenAccept(new Consumer<JSONObject>() {
                @Override
                public void accept(JSONObject dim) {
                    // 1. 补充维度信息
                    addDim(bean, dim);
                    // 2. 把 bean 输出
                    resultFuture.complete(Collections.singleton(bean));
                }
            });
    }
    
    @Override
    public void timeout(T input,
                        ResultFuture<T> resultFuture) throws Exception {
        System.out.println("异步超时异常:\n" +
                               "    1. 检查所有集群是否正常\n" +
                               "            hdfs hbase redis kafka\n" +
                               "    2. 再检测,用到的 6 张表是否都正常, 数据是否完整\n" +
                               "    \n" +
                               "    3. 加大超时时间");
        // 当异步超时的时候, 会自动调用这个方法
        super.timeout(input, resultFuture);
    }
}
